NexusPi Git Node
Commit 443700a629eee5c0fcbbdf9908a62a3dec6218e8
Parents : f547c49
Author : Ivan <e318cbc04468bd574db2b4523dddd710>
Signature : T66BB85Valid, signed by author
Date : 2026-08-19T16:01:46-05:00
feat: harden IdentityContext and DatabaseProvider with improved deferred management and integrity checks
Changes
13 files changed, 686 insertions(+), 106 deletions(-)
Diff
diff --git a/meshchatx.rsm b/meshchatx.rsm
index ae246b34..a7a3c769 100644
Binary files a/meshchatx.rsm and b/meshchatx.rsm differ
diff --git a/meshchatx/src/backend/database/__init__.py b/meshchatx/src/backend/database/__init__.py
index b0e236b6..0537fe1e 100644
--- a/meshchatx/src/backend/database/__init__.py
+++ b/meshchatx/src/backend/database/__init__.py
@@ -651,11 +651,16 @@ class Database:
actions.append({"step": "wal_checkpoint", "result": self._checkpoint_wal()})
integrity_rows = self.provider.integrity_check()
- integrity = [row[0] for row in integrity_rows] if integrity_rows else []
+ integrity = self._integrity_labels(integrity_rows)
actions.append({"step": "integrity_check", "result": integrity})
- self.provider.vacuum()
- self._tune_sqlite_pragmas()
+ integrity_ok = bool(integrity) and integrity[0] == "ok"
+ if integrity_ok:
+ self.provider.vacuum()
+ actions.append({"step": "vacuum", "result": "ok"})
+ self._tune_sqlite_pragmas()
+ else:
+ actions.append({"step": "vacuum", "result": "skipped"})
actions.append(
{
@@ -669,6 +674,16 @@ class Database:
"health": self.get_database_health_snapshot(),
}
+ @staticmethod
+ def _integrity_labels(rows) -> list:
+ labels = []
+ for row in rows or []:
+ if isinstance(row, dict):
+ labels.append(next(iter(row.values())))
+ else:
+ labels.append(row[0])
+ return labels
+
def _checkpoint_and_close(self):
try:
self._checkpoint_wal()
@@ -897,6 +912,14 @@ class Database:
result["current_stats"] = current_stats
return result
+ if current_stats.get("message_count", 0) < 0:
+ _log.warning(
+ "Backup message count unavailable, skipping rotation and baseline update",
+ )
+ result["count_unknown"] = True
+ result["current_stats"] = current_stats
+ return result
+
if max_count is not None and max_count > 0:
try:
backups = []
@@ -904,6 +927,7 @@ class Database:
if (
file.endswith(".zip")
and file.startswith("backup-")
+ and not file.startswith(PRE_MIGRATE_BACKUP_PREFIX)
and "SUSPICIOUS" not in file
):
full_path = os.path.join(default_dir, file)
@@ -1071,25 +1095,6 @@ class Database:
if os.path.exists(staged):
shutil.move(staged, paths[key])
- # Copy any remaining identity-storage files from the zip staging tree.
- for root, _dirs, files in os.walk(staging_dir, followlinks=False):
- for name in files:
- src = os.path.join(root, name)
- rel = os.path.relpath(src, staging_dir)
- if rel in {main_name, f"{main_name}-wal", f"{main_name}-shm"}:
- continue
- if name.endswith(("-wal", "-shm")) and name.startswith(
- os.path.splitext(main_name)[0],
- ):
- continue
- dest = os.path.join(target_dir, rel)
- dest_dir = os.path.dirname(dest)
- if dest_dir:
- os.makedirs(dest_dir, exist_ok=True)
- if os.path.islink(src):
- continue
- shutil.copy2(src, dest)
-
try:
self.initialize()
except Exception as exc:
@@ -1101,12 +1106,7 @@ class Database:
) from exc
self._tune_sqlite_pragmas()
integrity_rows = self.provider.integrity_check()
- integrity = []
- for row in integrity_rows or []:
- if isinstance(row, dict):
- integrity.append(next(iter(row.values())))
- else:
- integrity.append(row[0])
+ integrity = self._integrity_labels(integrity_rows)
if integrity and integrity[0] != "ok":
self.close_all()
self._restore_aside_files(aside_dir, paths)
@@ -1115,6 +1115,20 @@ class Database:
raise DatabaseRestoreError(
f"Restored backup failed integrity check: {integrity[0]!s}",
)
+ try:
+ self._copy_identity_storage_from_staging(
+ staging_dir,
+ target_dir,
+ main_name,
+ )
+ except Exception as exc:
+ self.close_all()
+ self._restore_aside_files(aside_dir, paths)
+ with suppress(Exception):
+ self.initialize()
+ raise DatabaseRestoreError(
+ f"Restored database but identity files failed to copy: {exc!s}",
+ ) from exc
finally:
shutil.rmtree(aside_dir, ignore_errors=True)
finally:
@@ -1126,6 +1140,70 @@ class Database:
"health": self.get_database_health_snapshot(),
}
+ @staticmethod
+ def _copy_identity_storage_from_staging(
+ staging_dir: str,
+ target_dir: str,
+ main_name: str,
+ ) -> None:
+ skip = {main_name, f"{main_name}-wal", f"{main_name}-shm"}
+ main_stem = os.path.splitext(main_name)[0]
+ extras_aside = tempfile.mkdtemp(
+ prefix=".meshchatx-extras-aside-",
+ dir=target_dir,
+ )
+ created: list[str] = []
+ try:
+ for root, _dirs, files in os.walk(staging_dir, followlinks=False):
+ for name in files:
+ src = os.path.join(root, name)
+ rel = os.path.relpath(src, staging_dir)
+ if rel in skip:
+ continue
+ if name.endswith(("-wal", "-shm")) and name.startswith(main_stem):
+ continue
+ if os.path.islink(src):
+ continue
+ dest = os.path.join(target_dir, rel)
+ dest_dir = os.path.dirname(dest)
+ if dest_dir:
+ os.makedirs(dest_dir, exist_ok=True)
+ if os.path.lexists(dest):
+ aside = os.path.join(extras_aside, rel)
+ aside_dir = os.path.dirname(aside)
+ if aside_dir:
+ os.makedirs(aside_dir, exist_ok=True)
+ shutil.move(dest, aside)
+ shutil.copy2(src, dest)
+ created.append(dest)
+ except Exception:
+ for dest in reversed(created):
+ with suppress(OSError):
+ if os.path.lexists(dest) and not os.path.isdir(dest):
+ os.remove(dest)
+ Database._restore_extras_aside(extras_aside, target_dir)
+ raise
+ finally:
+ shutil.rmtree(extras_aside, ignore_errors=True)
+
+ @staticmethod
+ def _restore_extras_aside(extras_aside: str, target_dir: str) -> None:
+ if not os.path.isdir(extras_aside):
+ return
+ for root, _dirs, files in os.walk(extras_aside, followlinks=False):
+ for name in files:
+ src = os.path.join(root, name)
+ rel = os.path.relpath(src, extras_aside)
+ dest = os.path.join(target_dir, rel)
+ dest_dir = os.path.dirname(dest)
+ if dest_dir:
+ os.makedirs(dest_dir, exist_ok=True)
+ if os.path.lexists(dest) and not os.path.isdir(dest):
+ with suppress(OSError):
+ os.remove(dest)
+ with suppress(OSError):
+ shutil.move(src, dest)
+
@staticmethod
def _restore_aside_files(aside_dir: str, paths: dict) -> None:
"""Put previously moved live DB files back after a failed restore."""
diff --git a/meshchatx/src/backend/database/auto_recover.py b/meshchatx/src/backend/database/auto_recover.py
index b97f1f4f..e392ec5b 100644
--- a/meshchatx/src/backend/database/auto_recover.py
+++ b/meshchatx/src/backend/database/auto_recover.py
@@ -78,11 +78,12 @@ def read_schema_version_from_db_path(db_path: str) -> int | None:
def infer_version_hint_from_backup_name(name: str) -> int | None:
+ """Return the schema version stored in a pre-migrate zip (the from version)."""
match = _PRE_MIGRATE_VERSION_RE.search(name)
if not match:
return None
try:
- return int(match.group(2))
+ return int(match.group(1))
except ValueError:
return None
@@ -171,15 +172,15 @@ def pick_compatible_backup(
for candidate in ordered:
probe = probe_backup_zip(candidate.path)
+ if probe.get("error"):
+ continue
version = probe.get("version")
if version is None:
version = infer_version_hint_from_backup_name(candidate.name)
if not schema_version_restorable(version, latest_schema_version):
continue
quick_check = probe.get("quick_check")
- if quick_check is not None and quick_check != "ok":
- continue
- if probe.get("error") and version is None:
+ if quick_check != "ok":
continue
return {
"name": candidate.name,
diff --git a/meshchatx/src/backend/database/messages.py b/meshchatx/src/backend/database/messages.py
index 6ee7acce..ea7db45f 100644
--- a/meshchatx/src/backend/database/messages.py
+++ b/meshchatx/src/backend/database/messages.py
@@ -49,6 +49,14 @@ _LXMF_OPTIONAL_UPSERT_FIELDS = frozenset(
},
)
_LXMF_EXPORT_ONLY_KEYS = frozenset({"id", "lxmf_icon"})
+_LXMF_KEEP_IF_INCOMING_BLANK = frozenset(
+ {
+ "content",
+ "title",
+ "fields",
+ "fields_meta",
+ },
+)
_LXMF_ATTACHMENT_FLAG_KEYS = (
"has_image",
"has_audio",
@@ -224,7 +232,9 @@ class MessageDAO:
"path_row_hash_hex",
)
]
- update_set = ", ".join([f"{f} = EXCLUDED.{f}" for f in update_fields])
+ update_set = ", ".join(
+ [self._lxmf_upsert_update_expr(field) for field in update_fields],
+ )
query = (
f"INSERT INTO lxmf_messages ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?) "
@@ -253,6 +263,21 @@ class MessageDAO:
if isinstance(peer_hash, str) and peer_hash.strip():
self.refresh_conversation_summary(peer_hash.strip())
+ @staticmethod
+ def _lxmf_upsert_update_expr(field: str) -> str:
+ if field not in _LXMF_KEEP_IF_INCOMING_BLANK:
+ return f"{field} = EXCLUDED.{field}"
+ if field in ("fields", "fields_meta"):
+ return (
+ f"{field} = CASE WHEN EXCLUDED.{field} IS NULL "
+ f"OR EXCLUDED.{field} = '' OR EXCLUDED.{field} = '{{}}' "
+ f"THEN lxmf_messages.{field} ELSE EXCLUDED.{field} END"
+ )
+ return (
+ f"{field} = CASE WHEN EXCLUDED.{field} IS NULL OR EXCLUDED.{field} = '' "
+ f"THEN lxmf_messages.{field} ELSE EXCLUDED.{field} END"
+ )
+
def refresh_conversation_summary(self, peer_hash):
"""Rebuild the materialized list row for one peer.
diff --git a/meshchatx/src/backend/database/provider.py b/meshchatx/src/backend/database/provider.py
index c592c982..80525d2e 100644
--- a/meshchatx/src/backend/database/provider.py
+++ b/meshchatx/src/backend/database/provider.py
@@ -3,7 +3,6 @@
import sqlite3
import sys
import threading
-import weakref
_SQLITE_CONNECT_KW = {}
if sys.version_info >= (3, 14):
@@ -15,12 +14,12 @@ _SQLITE_BUSY_TIMEOUT_MS = 5000
class DatabaseProvider:
_instance = None
_lock = threading.RLock()
- _all_locals = weakref.WeakSet()
def __init__(self, db_path=None):
self.db_path = db_path
self._local = threading.local()
- self._all_locals.add(self._local)
+ self._connections = set()
+ self._close_generation = 0
self._memory_connection = None
# Per-connection default. Worker threads opened via asyncio.to_thread
# never see Database._tune_sqlite_pragmas(), so this must be set here.
@@ -87,21 +86,33 @@ class DatabaseProvider:
self._configure_connection(self._memory_connection)
return self._memory_connection
- if not hasattr(self._local, "connection"):
+ local_gen = getattr(self._local, "generation", None)
+ if (
+ not hasattr(self._local, "connection")
+ or local_gen != self._close_generation
+ ):
if self.db_path is None:
msg = "db_path is required for database connections"
raise ValueError(msg)
- # isolation_level=None enables autocommit mode, letting us manage transactions manually
- self._local.connection = sqlite3.connect(
- self.db_path,
- timeout=30.0,
- check_same_thread=False,
- isolation_level=None,
- **_SQLITE_CONNECT_KW,
- )
- self._local.connection.row_factory = sqlite3.Row
- if self.db_path != ":memory:":
- self._configure_connection(self._local.connection)
+ with self._lock:
+ local_gen = getattr(self._local, "generation", None)
+ if (
+ hasattr(self._local, "connection")
+ and local_gen == self._close_generation
+ ):
+ return self._local.connection
+ conn = sqlite3.connect(
+ self.db_path,
+ timeout=30.0,
+ check_same_thread=False,
+ isolation_level=None,
+ **_SQLITE_CONNECT_KW,
+ )
+ conn.row_factory = sqlite3.Row
+ self._configure_connection(conn)
+ self._local.connection = conn
+ self._local.generation = self._close_generation
+ self._connections.add(conn)
return self._local.connection
def execute(self, query, params=None, commit=None):
@@ -194,15 +205,19 @@ class DatabaseProvider:
self._memory_connection = None
if hasattr(self._local, "connection"):
+ conn = self._local.connection
try:
- self.commit() # Ensure everything is saved
- self._local.connection.close()
+ self.commit()
+ conn.close()
except Exception:
pass
+ with self._lock:
+ self._connections.discard(conn)
del self._local.connection
def close_all(self):
with self._lock:
+ self._close_generation += 1
if self._memory_connection:
try:
self._memory_connection.commit()
@@ -211,14 +226,15 @@ class DatabaseProvider:
pass
self._memory_connection = None
- for loc in self._all_locals:
- if hasattr(loc, "connection"):
- try:
- loc.connection.commit()
- loc.connection.close()
- except Exception:
- pass
- del loc.connection
+ for conn in list(self._connections):
+ try:
+ conn.commit()
+ conn.close()
+ except Exception:
+ pass
+ self._connections.clear()
+ if hasattr(self._local, "connection"):
+ del self._local.connection
def vacuum(self):
# VACUUM cannot run inside a transaction
diff --git a/meshchatx/src/backend/identity_context.py b/meshchatx/src/backend/identity_context.py
index 3e0a0f16..aff63fa9 100644
--- a/meshchatx/src/backend/identity_context.py
+++ b/meshchatx/src/backend/identity_context.py
@@ -45,6 +45,8 @@ from meshchatx.src.backend.voicemail_manager import VoicemailManager
class IdentityContext:
+ DEFERRED_SETUP_TEARDOWN_WAIT_S = 30
+
def __init__(self, identity: RNS.Identity, app):
self.identity = identity
self.app = app
@@ -471,6 +473,25 @@ class IdentityContext:
def _deferred_still_active(self) -> bool:
return bool(self.running)
+ @staticmethod
+ def _discard_deferred_value(value) -> None:
+ if value is None:
+ return
+ for method_name in ("shutdown", "teardown", "stop", "cleanup"):
+ method = getattr(value, method_name, None)
+ if callable(method):
+ with contextlib.suppress(Exception):
+ method()
+ return
+
+ def _set_if_running(self, name: str, value) -> bool:
+ with self._deferred_setup_lock:
+ if not self.running:
+ self._discard_deferred_value(value)
+ return False
+ setattr(self, name, value)
+ return True
+
def _run_deferred_services_body(self):
if not self._deferred_still_active():
return
@@ -478,7 +499,7 @@ class IdentityContext:
try:
if not self._deferred_still_active():
return
- self.map_overlay_manager = MapOverlayManager(
+ overlay = MapOverlayManager(
self.config,
self.database,
self.storage_path,
@@ -486,11 +507,13 @@ class IdentityContext:
identity=self.identity,
reticulum=getattr(self.app, "reticulum", None),
)
+ if not self._set_if_running("map_overlay_manager", overlay):
+ return
try:
- self.map_overlay_manager.start_scheduler()
+ overlay.start_scheduler()
except Exception:
pass
- self.map_data_manager = MapDataManager(
+ data_mgr = MapDataManager(
self.config,
self.database,
self.storage_path,
@@ -499,8 +522,10 @@ class IdentityContext:
link_manager_getter=lambda: getattr(self.app, "rns_link_manager", None),
overlay_manager_getter=lambda: self.map_overlay_manager,
)
+ if not self._set_if_running("map_data_manager", data_mgr):
+ return
try:
- self.map_data_manager.start()
+ data_mgr.start()
except Exception as exc:
print(f"Failed to start map data manager: {exc}")
except Exception as exc:
@@ -516,83 +541,101 @@ class IdentityContext:
return
try:
- self.rncp_handler = RNCPHandler(
+ rncp = RNCPHandler(
reticulum_instance=getattr(self.app, "reticulum", None),
identity=self.identity,
storage_dir=self.app.storage_dir,
)
- self.rncp_handler.on_receive_completed = self._rncp_emit_receive_completed
+ rncp.on_receive_completed = self._rncp_emit_receive_completed
+ if not self._set_if_running("rncp_handler", rncp):
+ return
if not self._deferred_still_active():
return
- self.rns_filesync_handler = RnsFilesyncHandler(
+ filesync = RnsFilesyncHandler(
reticulum_instance=getattr(self.app, "reticulum", None),
identity=self.identity,
storage_dir=self.storage_path,
emit_callback=self._filesync_emit,
)
- self.rnsh_manager = RNSHManager(
+ if not self._set_if_running("rns_filesync_handler", filesync):
+ return
+ rnsh = RNSHManager(
storage_dir=self.storage_path,
reticulum_config_dir=getattr(self.app, "reticulum_config_dir", None),
)
- self.rnsh_manager.set_change_callback(
+ rnsh.set_change_callback(
lambda session: self.app.on_rnsh_change(session, context=self),
)
- self.rnsh_manager.set_output_callback(
+ rnsh.set_output_callback(
lambda session, chunk: self.app.on_rnsh_output(
session,
chunk,
context=self,
),
)
+ if not self._set_if_running("rnsh_manager", rnsh):
+ return
try:
- self.rnsh_manager.load()
+ rnsh.load()
except Exception as exc:
print(f"Failed to load RNSH sessions for {self.identity_hash}: {exc}")
if not self._deferred_still_active():
return
- self.rnx_manager = RNXManager(
+ rnx = RNXManager(
storage_dir=self.storage_path,
reticulum_config_dir=getattr(self.app, "reticulum_config_dir", None),
)
- self.rnx_manager.set_change_callback(
+ rnx.set_change_callback(
lambda session: self.app.on_rnx_change(session, context=self),
)
- self.rnx_manager.set_output_callback(
+ rnx.set_output_callback(
lambda session, chunk: self.app.on_rnx_output(
session,
chunk,
context=self,
),
)
+ if not self._set_if_running("rnx_manager", rnx):
+ return
try:
- self.rnx_manager.load()
+ rnx.load()
except Exception as exc:
print(f"Failed to load RNX sessions for {self.identity_hash}: {exc}")
- self.rnstatus_handler = RNStatusHandler(
+ status = RNStatusHandler(
reticulum_instance=getattr(self.app, "reticulum", None),
)
- self.rnpath_handler = RNPathHandler(
+ if not self._set_if_running("rnstatus_handler", status):
+ return
+ path_handler = RNPathHandler(
reticulum_instance=getattr(self.app, "reticulum", None),
)
- self.rnpath_trace_handler = RNPathTraceHandler(
+ if not self._set_if_running("rnpath_handler", path_handler):
+ return
+ trace = RNPathTraceHandler(
reticulum_instance=getattr(self.app, "reticulum", None),
identity=self.identity,
)
- self.rnprobe_handler = RNProbeHandler(
+ if not self._set_if_running("rnpath_trace_handler", trace):
+ return
+ probe = RNProbeHandler(
reticulum_instance=getattr(self.app, "reticulum", None),
identity=self.identity,
)
+ if not self._set_if_running("rnprobe_handler", probe):
+ return
libretranslate_url = self.config.libretranslate_url.get()
libretranslate_api_key = self.config.libretranslate_api_key.get()
- self.translator_handler = TranslatorHandler(
+ translator = TranslatorHandler(
libretranslate_url=libretranslate_url,
libretranslate_api_key=libretranslate_api_key,
translator_argos_enabled=self.config.translator_argos_enabled.get(),
translator_libretranslate_enabled=self.config.translator_libretranslate_enabled.get(),
)
+ if not self._set_if_running("translator_handler", translator):
+ return
- self.bot_handler = BotHandler(
+ bots = BotHandler(
identity_path=self.storage_path,
config_manager=self.config,
default_reticulum_config_dir=getattr(
@@ -601,8 +644,10 @@ class IdentityContext:
None,
),
)
+ if not self._set_if_running("bot_handler", bots):
+ return
try:
- self.bot_handler.restore_enabled_bots()
+ bots.restore_enabled_bots()
except Exception as exc:
print(f"Failed to restore bots: {exc}")
except Exception as exc:
@@ -614,7 +659,7 @@ class IdentityContext:
try:
rrc_enabled = self.config.rrc_enabled.get() if self.config else True
if rrc_enabled:
- self.rrc_manager = RRCManager(
+ rrc = RRCManager(
identity=self.identity,
storage_dir=self.storage_path,
get_nickname=lambda: (
@@ -623,41 +668,45 @@ class IdentityContext:
get_name_for_identity_hash=self._rrc_name_for_identity_hash,
database=self.database,
)
- self.rrc_manager.set_change_callback(
+ rrc.set_change_callback(
lambda hub: self.app.on_rrc_change(hub, context=self),
)
- self.rrc_manager.set_message_callback(
+ rrc.set_message_callback(
lambda hub, msg: self.app.on_rrc_message(hub, msg, context=self),
)
+ if not self._set_if_running("rrc_manager", rrc):
+ return
try:
- self.rrc_manager.load()
+ rrc.load()
except Exception as exc:
print(f"Failed to load RRC hubs for {self.identity_hash}: {exc}")
- self.rrc_server_manager = RRCServerManager(
+ server = RRCServerManager(
storage_dir=self.storage_path,
owner_identity=self.identity.hash,
)
- self.rrc_server_manager.set_change_callback(
+ server.set_change_callback(
lambda hub: self.app.on_rrc_server_change(hub, context=self),
)
- self.rrc_manager.set_server_manager(self.rrc_server_manager)
+ if not self._set_if_running("rrc_server_manager", server):
+ return
+ rrc.set_server_manager(server)
try:
- self.rrc_server_manager.load()
+ server.load()
except Exception as exc:
print(
f"Failed to load RRC hub servers for {self.identity_hash}: {exc}",
)
try:
- self.rrc_manager.connect_auto_reconnect_hubs()
+ rrc.connect_auto_reconnect_hubs()
except Exception as exc:
print(
f"Failed to auto-connect RRC hubs for {self.identity_hash}: {exc}",
)
else:
- self.rrc_manager = None
- self.rrc_server_manager = None
+ self._set_if_running("rrc_manager", None)
+ self._set_if_running("rrc_server_manager", None)
except Exception as exc:
print(f"Failed deferred RRC setup: {exc}")
@@ -860,11 +909,13 @@ class IdentityContext:
def teardown(self):
print(f"Tearing down Identity Context for {self.identity_hash}...")
- self.running = False
+ with self._deferred_setup_lock:
+ self.running = False
# Let an in-flight deferred setup notice running=False and exit before
# we null managers it may still be assigning.
finished = getattr(self, "_deferred_setup_finished", None)
- if finished is not None and not finished.wait(timeout=30):
+ wait_s = getattr(self, "DEFERRED_SETUP_TEARDOWN_WAIT_S", 30)
+ if finished is not None and not finished.wait(timeout=wait_s):
print(
f"Timed out waiting for deferred setup during teardown of {self.identity_hash}",
)
diff --git a/meshchatx/src/backend/rrc/manager.py b/meshchatx/src/backend/rrc/manager.py
index 6a28895f..f427c732 100644
--- a/meshchatx/src/backend/rrc/manager.py
+++ b/meshchatx/src/backend/rrc/manager.py
@@ -105,6 +105,7 @@ class RRCHub:
self.clean_last_removed = 0
self.available_rooms = {}
+ self.available_keyed_rooms = []
self._silent_list_pending = 0
self._silent_who_rooms = set()
@@ -1262,10 +1263,15 @@ class RRCHub:
if new_nick:
self.set_nick_override(new_nick)
- parsed = proto.parse_room_list_notice(body)
+ parsed = proto.parse_room_list_notice_details(body)
if parsed is not None:
with self._lock:
- self.available_rooms = parsed
+ self.available_rooms = {
+ name: info.get("topic") for name, info in parsed.items()
+ }
+ self.available_keyed_rooms = sorted(
+ name for name, info in parsed.items() if info.get("has_key")
+ )
silent = self._silent_list_pending > 0
if silent:
self._silent_list_pending -= 1
@@ -1507,6 +1513,7 @@ class RRCHub:
"total_unread": total_unread,
"mention_rooms": sorted(self.mention_rooms),
"available_rooms": dict(self.available_rooms),
+ "available_keyed_rooms": list(self.available_keyed_rooms),
"stored_key_rooms": stored_key_rooms,
"auto_reconnect": bool(self.auto_reconnect),
"auto_list": bool(self.auto_list),
diff --git a/meshchatx/src/backend/rrc/protocol.py b/meshchatx/src/backend/rrc/protocol.py
index 27b5ead4..3195cd50 100644
--- a/meshchatx/src/backend/rrc/protocol.py
+++ b/meshchatx/src/backend/rrc/protocol.py
@@ -240,8 +240,31 @@ def parse_who_notice(text):
return (room, entries)
-def parse_room_list_notice(text):
- """Parse a hub /list notice into {room: topic_or_None} or None."""
+# Optional [+k] after the room name in /list. The actual key is never listed.
+_LIST_KEYED_SUFFIX_RE = re.compile(r"^(?P<name>.*?)\s*\[\+k\]\s*$", re.IGNORECASE)
+
+
+def _split_list_room_name(name_part, *, strip_hash):
+ raw = name_part.strip()
+ if strip_hash:
+ raw = raw.lstrip("#")
+ has_key = False
+ matched = _LIST_KEYED_SUFFIX_RE.match(raw)
+ if matched:
+ raw = matched.group("name").strip()
+ if strip_hash:
+ raw = raw.lstrip("#")
+ has_key = True
+ return raw.lower(), has_key
+
+
+def parse_room_list_notice_details(text):
+ """Parse a hub /list notice into room metadata or None.
+
+ Values are dicts with topic (str or None) and has_key (bool). Optional
+ [+k] after the room name marks a keyed room and is stripped from the
+ name. The actual key is never present on the wire.
+ """
if not isinstance(text, str):
return None
stripped = text.strip()
@@ -255,19 +278,31 @@ def parse_room_list_notice(text):
s = line.strip()
if not s:
continue
+ topic = None
+ strip_hash = True
if s.endswith(" -"):
- name = s[:-2].strip().lstrip("#").lower()
- if name:
- rooms[name] = None
- continue
- if " - " in s:
- name, topic = s.split(" - ", 1)
- rooms[name.strip().lower()] = topic.strip() or None
+ name_part = s[:-2]
+ elif " - " in s:
+ name_part, topic_part = s.split(" - ", 1)
+ topic = topic_part.strip() or None
+ strip_hash = False
else:
- rooms[s.strip().lstrip("#").lower()] = None
+ name_part = s
+ name, has_key = _split_list_room_name(name_part, strip_hash=strip_hash)
+ if not name:
+ continue
+ rooms[name] = {"topic": topic, "has_key": has_key}
return rooms
+def parse_room_list_notice(text):
+ """Parse a hub /list notice into {room: topic_or_None} or None."""
+ details = parse_room_list_notice_details(text)
+ if details is None:
+ return None
+ return {name: info.get("topic") for name, info in details.items()}
+
+
class RRCMessage:
"""A single chat event (message, action, notice, or system line)."""
diff --git a/tests/backend/test_database_auto_recover.py b/tests/backend/test_database_auto_recover.py
index d779de43..6839434d 100644
--- a/tests/backend/test_database_auto_recover.py
+++ b/tests/backend/test_database_auto_recover.py
@@ -27,10 +27,25 @@ def test_schema_version_restorable_bounds():
def test_infer_version_hint_from_pre_migrate_name():
name = f"{PRE_MIGRATE_BACKUP_PREFIX}v52-to-v53-20260101-120000.zip"
- assert infer_version_hint_from_backup_name(name) == 53
+ assert infer_version_hint_from_backup_name(name) == 52
assert infer_version_hint_from_backup_name("backup-2026.zip") is None
+def test_pick_compatible_backup_skips_unprobed_pre_migrate_name(tmp_path):
+ temp_dir = str(tmp_path)
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ os.makedirs(backup_dir)
+ bogus = os.path.join(
+ backup_dir,
+ f"{PRE_MIGRATE_BACKUP_PREFIX}v52-to-v53-20260101-120000.zip",
+ )
+ with open(bogus, "wb") as handle:
+ handle.write(b"not-a-zip")
+
+ picked = pick_compatible_backup(temp_dir, DatabaseSchema.LATEST_VERSION)
+ assert picked is None
+
+
def test_pick_compatible_backup_skips_too_new_and_suspicious(tmp_path):
temp_dir = str(tmp_path)
db_path = os.path.join(temp_dir, "database.db")
diff --git a/tests/backend/test_database_lifecycle_safety.py b/tests/backend/test_database_lifecycle_safety.py
index 26b592b3..982e514a 100644
--- a/tests/backend/test_database_lifecycle_safety.py
+++ b/tests/backend/test_database_lifecycle_safety.py
@@ -4,6 +4,7 @@ import os
import shutil
import sqlite3
import tempfile
+import threading
import unittest
from unittest.mock import patch
@@ -51,6 +52,54 @@ def test_provider_path_switch_calls_close_all(temp_dir):
DatabaseProvider._instance.close_all()
+def test_provider_close_all_does_not_close_other_providers(temp_dir):
+ live_path = os.path.join(temp_dir, "live.db")
+ other_path = os.path.join(temp_dir, "other.db")
+ live = DatabaseProvider.get_instance(live_path)
+ live.execute("CREATE TABLE keep (id INTEGER PRIMARY KEY, val TEXT)")
+ live.execute("INSERT INTO keep (val) VALUES (?)", ("alive",))
+ live_conn = live.connection
+
+ other = DatabaseProvider(other_path)
+ other.execute("CREATE TABLE tmp (id INTEGER PRIMARY KEY)")
+ other.close_all()
+
+ row = live_conn.execute("SELECT val FROM keep").fetchone()
+ assert row is not None
+ assert row[0] == "alive"
+ assert live.connection is live_conn
+ live.close_all()
+
+
+def test_close_all_closes_worker_thread_connections(temp_dir):
+ db_path = os.path.join(temp_dir, "worker.db")
+ provider = DatabaseProvider(db_path)
+ provider.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, val TEXT)")
+ barrier = threading.Barrier(2)
+ held = {}
+
+ def worker():
+ provider.execute("INSERT INTO t (val) VALUES (?)", ("w",))
+ held["conn"] = provider.connection
+ barrier.wait()
+ barrier.wait()
+ try:
+ held["conn"].execute("SELECT 1")
+ held["still_open"] = True
+ except sqlite3.ProgrammingError:
+ held["still_open"] = False
+
+ thread = threading.Thread(target=worker)
+ thread.start()
+ barrier.wait()
+ provider.close_all()
+ barrier.wait()
+ thread.join(timeout=5)
+ assert held.get("still_open") is False
+ with pytest.raises(sqlite3.ProgrammingError, match="closed"):
+ held["conn"].execute("SELECT 1")
+
+
def test_restore_invokes_close_all_before_replace(temp_dir):
db_path = os.path.join(temp_dir, "live.db")
db = Database(db_path)
diff --git a/tests/backend/test_database_snapshots.py b/tests/backend/test_database_snapshots.py
index d0105300..a98e8e48 100644
--- a/tests/backend/test_database_snapshots.py
+++ b/tests/backend/test_database_snapshots.py
@@ -220,6 +220,36 @@ def test_backup_normal_rotation_and_baseline_update(temp_dir):
assert os.path.exists(os.path.join(backup_dir, "backup-baseline.json"))
+def test_backup_rotation_preserves_pre_migrate_zips(temp_dir):
+ import time
+
+ from meshchatx.src.backend.database import PRE_MIGRATE_BACKUP_PREFIX
+
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ os.makedirs(backup_dir, exist_ok=True)
+ pre_migrate_name = f"{PRE_MIGRATE_BACKUP_PREFIX}v1-to-v2-20000101-000000.zip"
+ pre_migrate_path = os.path.join(backup_dir, pre_migrate_name)
+ with open(pre_migrate_path, "wb") as handle:
+ handle.write(b"PK\x03\x04pre-migrate-keep")
+ old = time.time() - 120
+ os.utime(pre_migrate_path, (old, old))
+
+ db.backup_database(temp_dir, max_count=1)
+ time.sleep(1.1)
+ db.backup_database(temp_dir, max_count=1)
+
+ assert os.path.isfile(pre_migrate_path)
+ regular = [
+ name
+ for name in os.listdir(backup_dir)
+ if name.endswith(".zip") and not name.startswith(PRE_MIGRATE_BACKUP_PREFIX)
+ ]
+ assert len(regular) == 1
+
+
def test_backup_failure_does_not_remove_existing_backups(temp_dir):
from unittest.mock import patch
@@ -237,6 +267,44 @@ def test_backup_failure_does_not_remove_existing_backups(temp_dir):
assert len(still_there) == 1
+def test_run_database_recovery_skips_vacuum_when_integrity_fails(temp_dir):
+ from unittest.mock import patch
+
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ db.execute_sql("INSERT INTO config (key, value) VALUES (?, ?)", ("keep", "me"))
+
+ with patch.object(
+ db.provider,
+ "integrity_check",
+ return_value=[{"integrity_check": "tree corrupt"}],
+ ):
+ with patch.object(db.provider, "vacuum") as mock_vacuum:
+ result = db.run_database_recovery()
+ mock_vacuum.assert_not_called()
+
+ vacuum_steps = [step for step in result["actions"] if step.get("step") == "vacuum"]
+ assert vacuum_steps
+ assert vacuum_steps[0]["result"] == "skipped"
+ row = db.provider.fetchone("SELECT value FROM config WHERE key = ?", ("keep",))
+ assert row["value"] == "me"
+ db.close_all()
+
+
+def test_run_database_recovery_accepts_dict_integrity_rows(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ result = db.run_database_recovery()
+ integrity_steps = [
+ step for step in result["actions"] if step.get("step") == "integrity_check"
+ ]
+ assert integrity_steps
+ assert integrity_steps[0]["result"][0] == "ok"
+ db.close_all()
+
+
def test_check_db_health_at_open_no_baseline_ok(temp_dir):
db_path = os.path.join(temp_dir, "test.db")
db = Database(db_path)
@@ -510,6 +578,121 @@ def test_restore_includes_identity_rrc_and_history(temp_dir):
assert row["value"] == "before-backup"
+def test_failed_restore_does_not_overwrite_identity_sidecars(temp_dir):
+ from unittest.mock import patch
+
+ from meshchatx.src.backend.database import DatabaseRestoreError
+
+ identity_dir = os.path.join(temp_dir, "identities", "abc123")
+ os.makedirs(identity_dir, exist_ok=True)
+ db_path = os.path.join(identity_dir, "database.db")
+ identity_path = os.path.join(identity_dir, "identity")
+ history_dir = os.path.join(identity_dir, "rrc_history", "hub1")
+ os.makedirs(history_dir, exist_ok=True)
+ history_path = os.path.join(history_dir, "lobby.log")
+
+ with open(identity_path, "wb") as handle:
+ handle.write(b"BACKUP-KEY")
+ with open(history_path, "wb") as handle:
+ handle.write(b"backup-history")
+
+ db = Database(db_path)
+ db.initialize()
+ backup = db.backup_database(identity_dir)
+
+ with open(identity_path, "wb") as handle:
+ handle.write(b"LIVE-KEY")
+ with open(history_path, "wb") as handle:
+ handle.write(b"live-history")
+
+ with patch.object(Database, "initialize", side_effect=RuntimeError("open failed")):
+ with pytest.raises(DatabaseRestoreError):
+ db.restore_database(backup["path"])
+
+ with open(identity_path, "rb") as handle:
+ assert handle.read() == b"LIVE-KEY"
+ with open(history_path, "rb") as handle:
+ assert handle.read() == b"live-history"
+ db.close_all()
+
+
+def test_partial_sidecar_copy_failure_restores_live_identity_files(temp_dir):
+ from unittest.mock import patch
+
+ from meshchatx.src.backend.database import DatabaseRestoreError
+
+ identity_dir = os.path.join(temp_dir, "identities", "abc123")
+ os.makedirs(identity_dir, exist_ok=True)
+ db_path = os.path.join(identity_dir, "database.db")
+ identity_path = os.path.join(identity_dir, "identity")
+ hubs_path = os.path.join(identity_dir, "rrc_hubs")
+ history_dir = os.path.join(identity_dir, "rrc_history", "hub1")
+ os.makedirs(history_dir, exist_ok=True)
+ history_path = os.path.join(history_dir, "lobby.log")
+
+ with open(identity_path, "wb") as handle:
+ handle.write(b"BACKUP-KEY")
+ with open(hubs_path, "wb") as handle:
+ handle.write(b"backup-hubs")
+ with open(history_path, "wb") as handle:
+ handle.write(b"backup-history")
+
+ db = Database(db_path)
+ db.initialize()
+ backup = db.backup_database(identity_dir)
+
+ with open(identity_path, "wb") as handle:
+ handle.write(b"LIVE-KEY")
+ with open(hubs_path, "wb") as handle:
+ handle.write(b"live-hubs")
+ with open(history_path, "wb") as handle:
+ handle.write(b"live-history")
+
+ real_copy = shutil.copy2
+
+ def flaky_copy(src, dest, *args, **kwargs):
+ if os.path.basename(src) == "lobby.log":
+ raise OSError("disk full")
+ return real_copy(src, dest, *args, **kwargs)
+
+ with patch("meshchatx.src.backend.database.shutil.copy2", side_effect=flaky_copy):
+ with pytest.raises(DatabaseRestoreError):
+ db.restore_database(backup["path"])
+
+ with open(identity_path, "rb") as handle:
+ assert handle.read() == b"LIVE-KEY"
+ with open(hubs_path, "rb") as handle:
+ assert handle.read() == b"live-hubs"
+ with open(history_path, "rb") as handle:
+ assert handle.read() == b"live-history"
+ db.close_all()
+
+
+def test_backup_count_failure_does_not_rotate_existing_backups(temp_dir):
+ import time
+ from unittest.mock import patch
+
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ first = db.backup_database(temp_dir, max_count=1)
+ first_name = os.path.basename(first["path"])
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ time.sleep(1.1)
+
+ with patch.object(
+ db.messages,
+ "count_lxmf_messages",
+ side_effect=RuntimeError("count failed"),
+ ):
+ db.backup_database(temp_dir, max_count=1)
+
+ names = [name for name in os.listdir(backup_dir) if name.endswith(".zip")]
+ assert first_name in names
+ assert len(names) >= 2
+ db.close_all()
+
+
def test_pre_migration_backup_written_before_schema_upgrade(temp_dir):
from meshchatx.src.backend.database.schema import DatabaseSchema
diff --git a/tests/backend/test_lifecycle.py b/tests/backend/test_lifecycle.py
index 2cd0239c..0fd387b4 100644
--- a/tests/backend/test_lifecycle.py
+++ b/tests/backend/test_lifecycle.py
@@ -5,6 +5,7 @@ import os
import shutil
import sqlite3
import tempfile
+import threading
from unittest.mock import MagicMock, patch
import pytest
@@ -121,6 +122,59 @@ def test_identity_context_teardown_completeness():
assert context.community_interfaces_manager is None
+def test_teardown_timeout_does_not_leave_deferred_manager():
+ mock_identity = MagicMock(spec=RNS.Identity)
+ mock_identity.hash = b"test_hash_32_bytes_long_01234567"
+ mock_identity.get_private_key.return_value = b"mock_pk"
+ mock_app = MagicMock()
+ mock_app.storage_dir = tempfile.mkdtemp()
+ mock_app.emergency = False
+ mock_app.cleanup_rns_state_for_identity = MagicMock()
+
+ started = threading.Event()
+ release = threading.Event()
+ dummy = MagicMock()
+
+ with (
+ patch("meshchatx.src.backend.identity_context.Database"),
+ patch("meshchatx.src.backend.identity_context.ConfigManager"),
+ patch("meshchatx.src.backend.identity_context.create_lxmf_router"),
+ patch("meshchatx.src.backend.identity_context.IntegrityManager"),
+ patch("meshchatx.src.backend.identity_context.AutoPropagationManager"),
+ patch("RNS.Transport"),
+ ):
+ context = IdentityContext(mock_identity, mock_app)
+ context.start_background_threads = MagicMock()
+ context.register_announce_handlers = MagicMock()
+
+ def delayed_body():
+ started.set()
+ release.wait(timeout=5)
+ context._set_if_running("rrc_manager", dummy)
+
+ context._run_deferred_services_body = delayed_body
+ context.running = True
+ context.database = MagicMock()
+ context.config = MagicMock()
+
+ worker = threading.Thread(target=context.setup_deferred_services)
+ worker.start()
+ assert started.wait(timeout=5)
+
+ original_wait = context._deferred_setup_finished.wait
+
+ def short_wait(timeout=None):
+ return original_wait(timeout=0.05)
+
+ context._deferred_setup_finished.wait = short_wait
+ context.teardown()
+ release.set()
+ worker.join(timeout=5)
+
+ assert context.rrc_manager is None
+ dummy.shutdown.assert_called()
+
+
def test_identity_context_memory_leak():
"""Verify that IdentityContext can be garbage collected after teardown."""
mock_identity = MagicMock(spec=RNS.Identity)
diff --git a/tests/backend/test_message_dao_extended.py b/tests/backend/test_message_dao_extended.py
index f7efe59f..24fad45a 100644
--- a/tests/backend/test_message_dao_extended.py
+++ b/tests/backend/test_message_dao_extended.py
@@ -228,3 +228,69 @@ def test_set_lxmf_message_path_at_send_if_unset(message_dao, mock_provider):
assert params[0] == 2
assert params[1] == "UDP Interface"
assert params[3] == "deadbeef"
+
+
+def test_upsert_empty_content_does_not_clobber_stored_body(real_db):
+ msg_hash = "a" * 32
+ _insert_message(real_db, msg_hash, is_incoming=1, state="delivered")
+ real_db.messages.upsert_lxmf_message(
+ {
+ "hash": msg_hash,
+ "source_hash": "b" * 32,
+ "destination_hash": "b" * 32,
+ "peer_hash": "b" * 32,
+ "state": "failed",
+ "progress": 0.0,
+ "is_incoming": 1,
+ "method": "direct",
+ "delivery_attempts": 3,
+ "next_delivery_attempt_at": None,
+ "title": "",
+ "content": "",
+ "fields": "{}",
+ "timestamp": time.time(),
+ "rssi": None,
+ "snr": None,
+ "quality": None,
+ "is_spam": 0,
+ "reply_to_hash": None,
+ "attachments_stripped": 0,
+ },
+ )
+ row = real_db.messages.get_lxmf_message_by_hash(msg_hash)
+ assert row["content"] == "c"
+ assert row["title"] == "t"
+ assert row["state"] == "failed"
+ assert row["delivery_attempts"] == 3
+
+
+def test_upsert_nonempty_content_still_replaces(real_db):
+ msg_hash = "c" * 32
+ _insert_message(real_db, msg_hash, is_incoming=1, state="delivered")
+ real_db.messages.upsert_lxmf_message(
+ {
+ "hash": msg_hash,
+ "source_hash": "b" * 32,
+ "destination_hash": "b" * 32,
+ "peer_hash": "b" * 32,
+ "state": "delivered",
+ "progress": 1.0,
+ "is_incoming": 1,
+ "method": "direct",
+ "delivery_attempts": 1,
+ "next_delivery_attempt_at": None,
+ "title": "new-title",
+ "content": "new-body",
+ "fields": '{"k":1}',
+ "timestamp": time.time(),
+ "rssi": None,
+ "snr": None,
+ "quality": None,
+ "is_spam": 0,
+ "reply_to_hash": None,
+ "attachments_stripped": 0,
+ },
+ )
+ row = real_db.messages.get_lxmf_message_by_hash(msg_hash)
+ assert row["content"] == "new-body"
+ assert row["title"] == "new-title"
Served by rngit 1.5.0 - Generated in 0.06s